package com.watson.kafka.lg.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;

/**
 * @author watson
 * @date 2021/7/29 9:22
 * @Description: TODO
 **/
@Configuration
public class KafkaInitialConfiguration {
    // 创建一个名为testtopic的Topic并设置分区数为8，分区副本数为2
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("testtopic", 8, (short) 2);
    }

    // 如果要修改分区数，只需修改配置值重启项目即可
    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
//    @Bean
//    public NewTopic updateTopic() {
//        return new NewTopic("testtopic",10, (short) 2 );
//    }


    /*
     * 通过异常处理器，我们可以处理consumer在消费时发生的异常。
     * 新建一个 ConsumerAwareListenerErrorHandler 类型的异常处理方法，用@Bean注入，
     * BeanName默认就是方法名，然后我们将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面，
     * 当监听抛出异常的时候，则会自动调用异常处理器
     * @return
     */


    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
    //@KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {

        return (message, exception, consumer) -> {
            System.out.println("消费异常：" + message.getPayload());
            return null;
        };
    }

    @Autowired
    ConsumerFactory consumerFactory;

    /**
     * 消息过滤器可以在消息抵达consumer之前被拦截，在实际应用中，我们可以根据自己的业务逻辑，筛选出需要的信息再交由KafkaListener处理，
     * 不需要的消息则过滤掉。
     * <p>
     * 配置消息过滤只需要为 监听器工厂 配置一个RecordFilterStrategy（消息过滤策略），返回true的时候消息将会被抛弃，
     * 返回false时，消息能正常抵达监听容器。
     *
     * @return
     */

    // 消息过滤器
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        // 消息过滤策略
        factory.setRecordFilterStrategy(consumerRecord -> {
//            if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
            return false;
//            }
            //返回true消息则被过滤
//            return true;
        });
        return factory;
    }


}
